#include "CanReaper.h"
#include "CanReader.h"
#include "utils.hpp"
#include "LoggerConfigs.h"

#include <fcntl.h>
#include <unistd.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/ioctl.h>
#include <linux/can.h>
#include <net/if.h>
#include <cstring>
#include <ranges>
#include <format>

#include "LoggerConfigs.h"
#include "Logger.h"


namespace canpp {
    using std::lock_guard;
    using std::ranges::views::keys;
    using std::ranges::views::values;
    using std::format;
    using std::vformat;
    using std::format_args;
    using std::make_format_args;
    using std::async;
    using std::move;
    string LOG_DIR = string(CANPP_LOG_DIR) + "/CanReaper/CanReaper";

    /* -------------- singleton -------------- */

    CanReaper &CanReaper::getInstance() {
        static CanReaper instance;
        return instance;
    }

    /* -------------- ///////// -------------- */


    void CanReaper::bindReader(CanReader *reader, const string &canInterface) {
        int interface_fd = -1;
        for (const auto &[fd, interface]: interfaces) {
            if (interface == canInterface) {
                interface_fd = fd;
                break;
            }
        }
        if (interface_fd == -1) {
            // new interface
            interface_fd = open_interface(canInterface);
            interfaces.emplace(interface_fd, canInterface);
        }
        readers.insert({interface_fd, reader});

        logger.info(format("Successfully bind Reader ID {}", reader->get_id()));
    }

    void CanReaper::removeReader(const CanReader *reader) {
        for (auto it = readers.begin(); it != readers.end();) {
            if (it->second == reader) {
                it = readers.erase(it);
            } else {
                ++it;
            }
        }
        // Check if the interface has no more readers and clean up
        for (auto it = interfaces.begin(); it != interfaces.end();) {
            if (auto [fst, snd] = readers.equal_range(it->first); fst == snd) {
                close(it->first);
                it = interfaces.erase(it);
            } else {
                ++it;
            }
        }
        logger.info(format("Successfully removed Reader ID {}", reader->get_id()));
    }

    int CanReaper::createOrGetInterfaceFd(const string &canInterface) {
        for (const auto &[fd, interface]: interfaces) {
            if (interface == canInterface) return fd;
        }
        int new_fd = open_can_socket(canInterface);
        interfaces.emplace(new_fd, canInterface);
        return new_fd;
    }

    int CanReaper::open_can_socket(const string &canInterface) const {
        const int sock_fd = socket(PF_CAN, SOCK_RAW, CAN_RAW);
        if (sock_fd == -1) {
            LOG_ERROR_AND_QUIT(logger, "Failed to create CAN socket");
        }

        ifreq ifr{};
        std::strncpy(ifr.ifr_name, canInterface.c_str(), IFNAMSIZ - 1);
        ifr.ifr_name[IFNAMSIZ - 1] = '\0';
        if (ioctl(sock_fd, SIOCGIFINDEX, &ifr) == -1) {
            close(sock_fd);
            LOG_ERROR_AND_QUIT(logger, "Failed to get CAN interface index");
        }

        sockaddr_can addr{};
        addr.can_family = AF_CAN;
        addr.can_ifindex = ifr.ifr_ifindex;
        if (bind(sock_fd, reinterpret_cast<sockaddr *>(&addr), sizeof(addr)) == -1) {
            close(sock_fd);
            LOG_ERROR_AND_QUIT(logger, "Failed to bind CAN socket");
        }

        const int flags = fcntl(sock_fd, F_GETFL, 0);
        fcntl(sock_fd, F_SETFL, flags | O_NONBLOCK);

        return sock_fd;
    }

    int CanReaper::open_interface(const string &canInterface) const {
        const int sock_fd = open_can_socket(canInterface);
        // flush linux buffer by nonblock reading
        can_frame frame{};
        while (true) {
            if (const ssize_t nbytes = read(sock_fd, &frame, sizeof(struct can_frame)); nbytes > 0) {
            } else if (nbytes == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
                // finish flushing
                break;
            } else {
                LOG_ERROR_AND_QUIT(logger, "Failed to flush linux buffer: " + string(strerror(errno)));
            }
        }

        epoll_event ev{};
        ev.events = EPOLLIN;
        ev.data.fd = sock_fd;
        if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, sock_fd, &ev) == -1) {
            LOG_ERROR_AND_QUIT(logger, "Failed to EPOLL_CTL_ADD new CAN sock: " + string(strerror(errno)));
        }

        // notify epoll_wait to add the new sock, prevent blocking
        constexpr uint64_t u = 1;
        if (write(new_sock_event_fd, &u, sizeof(uint64_t)) == -1) {
            LOG_ERROR_AND_QUIT(logger, "Failed to write new_sock_event_fd: " + string(strerror(errno)));
        };

        logger.info(format("Successfully opened CAN interface: {}", canInterface));

        return sock_fd;
    }

    void CanReaper::reap_process(const std::stop_token &stop_token) {
        epoll_event events[CAN_REAPER_MAX_EVENT]{};
        while (true) {
            int nfds = epoll_wait(epoll_fd, events, CAN_REAPER_MAX_EVENT, -1);
            if (nfds == -1) {
                if (stop_token.stop_requested()) {
                    return;
                }
                LOG_ERROR_AND_QUIT(logger, "Failed to epoll_wait: " + string(strerror(errno)));
            }

            for (int i = 0; i < nfds; ++i) {
                if (events[i].data.fd == new_sock_event_fd) {
                    if (stop_token.stop_requested()) {
                        return;
                    }
                    uint64_t u;
                    if (read(new_sock_event_fd, &u, sizeof(uint64_t)) < 0) {
                        LOG_ERROR_AND_QUIT(logger, "Failed to read CAN sock: " + string(strerror(errno)));
                    };
                } else if (events[i].events & EPOLLIN) {
                    can_frame frame{};
                    if (const int fd = events[i].data.fd; read(fd, &frame, sizeof(frame)) > 0) {
                        logger.info(
                            format("Message from {}: {}",
                                   interfaces[fd], format_frame_data(frame))
                        );
                        logger.debug(format("Message fd: {}", fd));
                        for (const auto &[first, last] = readers.equal_range(fd); const auto &[_, reader]:
                             std::ranges::subrange(first, last)) {
                            pool.detach_task([&] {
                                reader->pushMessage(move(can_frame(frame)));
                            });
                        }
                    }
                }
            }
        }
    }

    void exit_handler() { CanReaper::getInstance().cleanUp(); };

    CanReaper::CanReaper(): logger([] {
        return log::create_log("CanReaper", vformat(reaperLoggerConfig
                                                    , format_args(make_format_args(LOG_DIR))));
    }()) {
        // init epoll
        epoll_fd = epoll_create(256);
        if (epoll_fd == -1) {
            LOG_ERROR_AND_QUIT(logger, "Failed to create epoll file descriptor: " + string(strerror(errno)));
        }

        // init epoll new_sock_event_fd
        new_sock_event_fd = eventfd(0, 0);
        if (new_sock_event_fd == -1) {
            LOG_ERROR_AND_QUIT(logger,
                               "Failed to create new_sock_event file descriptor: " + string(strerror(errno)));
        }

        epoll_event ev{};
        ev.events = EPOLLIN;
        ev.data.fd = new_sock_event_fd;
        if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, new_sock_event_fd, &ev) == -1) {
            LOG_ERROR_AND_QUIT(logger, "Failed to EPOLL_CTL_ADD new_sock_event: " + string(strerror(errno)));
        }
        reap_thread = jthread([this](const std::stop_token &stop_token) {
            try {
                reap_process(stop_token);
                logger.force_flush();
                log::force_flush_all_logs();
                log::uninit();
                pool.wait();
            } catch (const std::exception_ptr &e) {
                std::rethrow_exception(e);
            }
        });
        // atexit(exit_handler);
    }


    void CanReaper::cleanUp() {
        reap_thread.request_stop();
        // wake up epoll_wait
        constexpr uint64_t u = 1;
        if (write(new_sock_event_fd, &u, sizeof(uint64_t)) == -1) {
            LOG_ERROR_AND_QUIT(logger, "Failed to write new_sock_event_fd: " + string(strerror(errno)));
        }
        reap_thread.join();
        for (const auto fd: keys(interfaces)) {
            close(fd);
        }
        close(epoll_fd);
        close(new_sock_event_fd);
    }

    CanReaper::~CanReaper() {
        cleanUp();
    }
} // canpp

